package org.hashdb.client.net.bio;

import com.fasterxml.jackson.core.JsonProcessingException;
import org.hashdb.client.config.AppConfig;
import org.hashdb.client.console.AppConsoleReader;
import org.hashdb.client.manager.AppContext;
import org.hashdb.client.net.bio.get.ActAuthenticationMessage;
import org.hashdb.client.net.bio.get.ErrorMessage;
import org.hashdb.client.net.bio.get.HeartbeatMessage;
import org.hashdb.client.net.bio.msg.Message;
import org.hashdb.client.net.bio.send.ActHeartbeatMessage;
import org.hashdb.client.net.bio.send.AuthenticationMessage;
import org.hashdb.client.net.bio.send.CloseMessage;
import org.hashdb.client.util.AsyncService;
import org.hashdb.client.util.JsonService;
import org.hashdb.client.util.event.EventEngin;

import java.io.IOException;
import java.io.PrintWriter;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * Date: 2023/12/2 16:30
 *
 * @author huanyuMake-pecdle
 * @version 0.0.1
 */
public class MessageChannel implements AutoCloseable {
    private final SocketChannel channel;

    private final ByteBuffer readBuffer = ByteBuffer.allocate(1024);

    private final BlockingQueue<Message> waitForConsumeMsgQueue = new LinkedBlockingQueue<>();
    private final BlockingQueue<Message> writeQueue = new LinkedBlockingQueue<>();

    private final MessageConsumerChain chain = new MessageConsumerChain();

    private final AppContext appContext;

    private record ResultWrapper(Object result) {
        public static ResultWrapper NULL = new ResultWrapper(null);
    }

    private final BlockingQueue<ResultWrapper> messageConsumeResultQueue = new LinkedBlockingQueue<>();

    private final CompletableFuture<?> messageReader;

    private final CompletableFuture<?> messageSender;

    private final CompletableFuture<?> messageConsumerDispatcher;

    private long lastGetTime;

    private int failAliveCheckCount = 0;

    private ScheduledFuture<?> aliveChecker;

    protected volatile boolean closed = false;

    protected AppConsoleReader consoleReader;

    protected PrintWriter stackWriter;

    protected final Thread shutdownHook;

    protected final EventEngin eventEngin = new EventEngin(m -> {
    });

    public static MessageChannel open(AppContext ctx) throws IOException {
        return new MessageChannel(ctx);
    }

    @SuppressWarnings("InfiniteLoopStatement")
    protected MessageChannel(AppContext appContext) throws IOException {
        this.channel = appContext.activeConnection();
        this.consoleReader = appContext.reader();
        this.appContext = appContext;
        stackWriter = new PrintWriter(consoleReader.getOutput());
        messageReader = AsyncService.submit(() -> {
            while (true) {
                try {
                    Message msg = get(true);
                    waitForConsumeMsgQueue.add(msg);
                } catch (AsynchronousCloseException e) {
                    // 其它线程关闭这个连接
                    return;
                } catch (IOException e) {
                    consoleReader.printStackTrace(e);
                    ErrorMessage errorMessage = new ErrorMessage();
                    errorMessage.setError(e.getClass().getSimpleName());
                    errorMessage.setCause(e.getMessage());
                    waitForConsumeMsgQueue.add(errorMessage);
                    consoleReader.errorLn("message reader stop to service. cause: " + e.getMessage());
                    return;
                } catch (IllegalStateException e) {
                    if (!e.getMessage().matches(".*no data.*")) {
                        consoleReader.errorLn("message reader throw " + e.getMessage());
                        throw e;
                    }
                }
            }
        });
        messageSender = AsyncService.submit(() -> {
            while (true) {
                try {
                    Message msg = writeQueue.take();
                    send0(msg);
                } catch (IOException | InterruptedException e) {
                    consoleReader.printStackTrace(e);
                    consoleReader.errorLn("message sender stop to service. cause: " + e.getMessage());
                    return;
                } catch (Exception e) {
                    consoleReader.errorLn("message sender throw " + e.getMessage());
                }
            }
        });
        messageConsumerDispatcher = AsyncService.submit(() -> {
            while (true) {
                Message msg;
                try {
                    msg = waitForConsumeMsgQueue.take();
                } catch (InterruptedException e) {
                    consoleReader.printStackTrace(e);
                    return;
                }
                Object result = null;
                try {
                    if (msg.isShouldConsume()) {
                        result = chain.consume(msg);
                    } else {
                        result = msg;
                    }
                } catch (Exception e) {
                    consoleReader.printStackTrace(e);
                }
                messageConsumeResultQueue.add(result == null ? ResultWrapper.NULL : new ResultWrapper(result));
            }
        });
        shutdownHook = new Thread(() -> {
            try {
                internalClose();
            } catch (IOException e) {
                consoleReader.printStackTrace(e);
                throw new RuntimeException(e);
            }
        });
        Runtime.getRuntime().addShutdownHook(shutdownHook);
        addDefaultMessageConsumer();
    }

    protected void addDefaultMessageConsumer() {
        // 确认服务器存活
        chain.add((msg, chain) -> {
            lastGetTime = System.currentTimeMillis();
            failAliveCheckCount = 0;
            return chain.next();
        });
        // 验证身份
        chain.add((msg, chain) -> {
            if (!(msg instanceof ActAuthenticationMessage act)) {
                if (appContext.getUser() != null) {
                    return chain.next();
                }
                if (msg instanceof CloseMessage) {
                    return chain.next();
                }
                if (msg instanceof ErrorMessage) {
                    return chain.next();
                }
                throw new RuntimeException("expected actAuth msg but get " + JsonService.toString(msg));
            }
            if (!act.isSuccess()) {
                try {
                    consoleReader.println("[AUTH_ERROR] " + JsonService.parse(act.getData()));
                } catch (JsonProcessingException e) {
                    consoleReader.printStackTrace(e);
                }
                return act;
            }
            consoleReader.messageLn(act.getData());
//            consoleReader.messageLn("publish authPass");
            eventEngin.publish("authPass", act.getUser());
            startCheckAlive();
            return act;
        });
        // 处理heartbeat
        chain.add((msg, chain) -> {
            if (!(msg instanceof HeartbeatMessage heartbeatMessage)) {
                return chain.next();
            }
            try {
                send(new ActHeartbeatMessage(heartbeatMessage));
            } catch (ClosedChannelException e) {
                consoleReader.printStackTrace(e);
                throw new RuntimeException(e);
            }
            return null;
        });
        chain.add((msg, chain) -> {
            if (!(msg instanceof CloseMessage closeMessage)) {
                return chain.next();
            }
            consoleReader.errorLn("server close connection. cause: '" + closeMessage.getData() + "'");
            try {
                internalClose();
            } catch (IOException e) {
                consoleReader.printStackTrace(e);
            }
            return null;
        });
    }

    public void authWithPassword(String username, String password) throws ClosedChannelException {
        Objects.requireNonNull(username);
        send(AuthenticationMessage.usePasswordAuth(username, password)).join();
    }

    public void auth() throws ClosedChannelException {
        var act = (ActAuthenticationMessage) send(new AuthenticationMessage()).join();
    }

    public boolean isConnected() {
        return !closed && channel.isConnected();
    }

    public synchronized void internalClose() throws IOException {
        if (closed) {
            return;
        }
        eventEngin.publish("closing");
        try {
            if (channel.isConnected()) {
                send0(new CloseMessage());
            }
        } catch (IOException e) {
            consoleReader.printStackTrace(e);
        }
        if (!Thread.currentThread().equals(shutdownHook)) {
            Runtime.getRuntime().removeShutdownHook(shutdownHook);
        }
        messageReader.cancel(true);
        messageSender.cancel(true);
        messageConsumerDispatcher.cancel(true);
        if (aliveChecker != null) {
            aliveChecker.cancel(true);
            aliveChecker = null;
        }
        try {
            channel.close();
        } catch (IOException e) {
            consoleReader.printStackTrace(e);
        }

        closed = true;
        eventEngin.publish("closed");
    }

    @Override
    public synchronized void close() throws IOException {
        internalClose();
    }

    protected Message get(boolean clear) throws IOException {
        if (closed || !channel.isConnected()) {
            throw new ClosedChannelException();
        }
        try {
//            readBuffer.mark();
            int readCount;
            int expectReadCount = readBuffer.position();
            while ((readCount = channel.read(readBuffer)) <= expectReadCount) {
                if (readCount == -1) {
                    throw new IllegalStateException("no data can read in read buffer");
                }
                expectReadCount -= readCount;
            }
        } catch (SocketException e) {
            consoleReader.printStackTrace(e);
            internalClose();
            throw e;
        } catch (IOException e) {
            internalClose();
            throw e;
        }
        try {
            Message result = JsonService.parse(new String(Arrays.copyOf(readBuffer.array(), readBuffer.position())), Message.class);
            if (clear) {
                readBuffer.clear();
            } else {
                readBuffer.reset();
            }
            consoleReader.messageLn("get msg: " + result);
//            System.out.println("get msg: " + result);
            return result;
        } catch (JsonProcessingException e) {
            if (clear) {
                readBuffer.clear();
            } else {
                readBuffer.reset();
            }
            consoleReader.printStackTrace(e);
            throw new RuntimeException("unexpected message type 'null'");
        }
    }

    public CompletableFuture<?> send(Message toSend) throws ClosedChannelException {
        writeQueue.add(toSend);
        return AsyncService.submit(() -> {
            try {
                return messageConsumeResultQueue.take().result();
            } catch (InterruptedException e) {
                consoleReader.printStackTrace(e);
                throw new RuntimeException(e);
            }
        });
    }

    public CompletableFuture<?> send(Message toSend, boolean shouldBeConsumed) throws ClosedChannelException {
        toSend.setShouldConsume(shouldBeConsumed);
        return send(toSend);
    }

    public CompletableFuture<?> send(Message toSend, long timeoutMillis) throws ClosedChannelException {
        writeQueue.add(toSend);
        ScheduledFuture<?>[] timeout = {null};
        var result = AsyncService.submit(() -> {
            try {
                var res = messageConsumeResultQueue.take().result();
                if (timeout[0] != null) {
                    timeout[0].cancel(true);
                }
                return res;
            } catch (InterruptedException e) {
                consoleReader.printStackTrace(e);
                throw new RuntimeException(e);
            }
        });
        timeout[0] = AsyncService.setTimeout(() -> {
            result.completeExceptionally(new TimeoutException("out of " + timeoutMillis + " ms"));
        }, timeoutMillis);
        return result;
    }

    protected void send0(Message toSend) throws IOException {
        if (closed || !channel.isConnected()) {
            throw new ClosedChannelException();
        }
        String data = JsonService.toString(toSend);
//        consoleReader.messageLn("send msg: " + data);
        try {
            channel.write(ByteBuffer.wrap(data.getBytes()));
        } catch (SocketException e) {
            consoleReader.printStackTrace(e);
            throw e;
        }
    }

    public void addMessageConsumer(MessageConsumer consumer) {
        chain.add(consumer);
    }

    protected synchronized void startCheckAlive() {
        long heartbeatInterval = AppConfig.getInstance().getHeartbeatInterval();
        int timeoutRetry = AppConfig.getInstance().getTimeoutRetry();
        if (aliveChecker != null) {
            aliveChecker.cancel(true);
        }
        aliveChecker = AsyncService.setInterval(() -> {
            if (System.currentTimeMillis() - lastGetTime <= heartbeatInterval) {
                failAliveCheckCount = 0;
                return;
            }
            if (failAliveCheckCount++ < timeoutRetry) {
                return;
            }
            consoleReader.errorLn("can not connect to server " + AppConfig.getInstance().getServer());
            try {
                internalClose();
            } catch (IOException e) {
                consoleReader.printStackTrace(e);
                throw new RuntimeException(e);
            }
        }, heartbeatInterval);
    }

    public void onClosed(Runnable cb) {
        eventEngin.register("closed", m -> cb.run());
    }

    public void onClosing(Runnable cb) {
        eventEngin.register("closing", m -> cb.run());
    }

    public void onAuthPass(Consumer<String> cb) {
        eventEngin.register("authPass", username_ -> {
            if (!(username_ instanceof String username)) {
                throw new IllegalArgumentException();
            }
            cb.accept(username);
        });
    }
}
